package com.chenwc.rocketmq.example;

import com.chenwc.rocketmq.prop.RocketMqConnectionProperties;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PushConsumerExample {
    private static final Logger logger = LoggerFactory.getLogger(PushConsumerExample.class);

    private PushConsumerExample() {
    }

    public static void main(String[] args) {
        logger.info("Consumer start");
        RocketMqConnectionProperties properties = new RocketMqConnectionProperties();
        String topic = "TopicTest";
        DefaultMQPushConsumer consumer;
        try {
            // 配置 AccessKey 和 SecretKey
            SessionCredentials sessionCredentials = new SessionCredentials(properties.getAccessKey(), properties.getSecretKey());
            RPCHook aclHook = new AclClientRPCHook(sessionCredentials);
            // 创建一个Consumer，并开启
            consumer = new DefaultMQPushConsumer("ConsumerGroup", aclHook);
            consumer.setNamesrvAddr(properties.getNamesrvAddr());

            // 从哪里开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            // 订阅Topic
            consumer.subscribe(topic, "*");
            // 注册回调接口
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                for (MessageExt msg : msgs) {
                    logger.info("Received message: {}", new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            // 广播模式
            consumer.setMessageModel(MessageModel.BROADCASTING);
            // 集群模式
            // consumer.setMessageModel(MessageModel.CLUSTERING);
            consumer.start();
        } catch (Exception e) {
            logger.error("Consumer start error", e);
        }
    }
}
